package com.apps.sdses.flink141.sql;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/9/6
 * desc:
 **/
public class MySQL2HIve {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        System.setProperty("hadoop.home.dir", "D:\\env\\hadoop-2.8.3");
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id INT,\n" +
                "    mode STRING,\n" +
                "    voltage FLOAT,\n" +
                "    resistance FLOAT,\n" +
                "    temperature FLOAT,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "   'connector' = 'jdbc',\n" +
                "   'url' = 'jdbc:mysql://192.168.102.155:33066/db_datacube_test',\n" +
                "   'driver' = 'com.mysql.cj.jdbc.Driver',\n" +
                "   'username' = 'datacube_test',\n" +
                "   'password' = 'Gbd#XERA@19d1ZtD',\n" +
                "   'table-name' = 'water_sensor_view'\n" +
                ")");


//        HiveCatalog hiveCatalog = new HiveCatalog("hive_test", "default", "src/main/resources/hive-conf");
//        tableEnv.registerCatalog("hive_test", hiveCatalog);
//        tableEnv.useCatalog();

//        String tableFullName = "`" + hiveCatalog.getName() + "`.`" + hiveCatalog.getDefaultDatabase() + "`.`test_ods_products`";
//        String tableFullName = "`hive_test`.`default`.`test_ods_products`";

//        tableEnv.executeSql("insert OVERWRITE  " + tableFullName + "  select *, DATE_FORMAT(LOCALTIMESTAMP, 'yyyyMMdd') from sourceTable");

        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print') LIKE sourceTable (EXCLUDING ALL)");
        tableEnv.executeSql("insert into sinkTable select * from sourceTable");

    }
}
